package com.mxnavi5.example.DataStream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class MyKeyedTest {
    private static final OutputTag<Integer> outputTag = new OutputTag<Integer>("OutputTag") {
    };

    public static void main(String[] args) throws Exception {
        //    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
//        env.setParallelism(3);
        env.disableOperatorChaining();


        //keyed 交换数据
        //完全不设置并行度
        SingleOutputStreamOperator returns1 = env.addSource(new UdfSourceTest()).returns(Integer.class);
        SingleOutputStreamOperator<Tuple2<Integer, Integer>> returns = returns1
                .map(x -> Tuple2.of(x, x))
                .returns(TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
                }));
        returns.print("re");
        returns.uid("flinkKeyedTest")
                .keyBy(1)
                .map(x -> x)
                .returns(TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {}))
                .print("key:");


        //算子设置并行度
//        SingleOutputStreamOperator returns1 = env.addSource(new UdfSourceTest()).returns(Integer.class);
//        SingleOutputStreamOperator<Tuple2<Integer, Integer>> returns = returns1
//                .map(x -> Tuple2.of(x, x)).setParallelism(3)
//                .returns(TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
//                }));
//        returns.print("re");
//        returns.uid("flinkKeyedTest")
//                .keyBy(1)
//                .map(x -> x).setParallelism(2)
//                .returns(TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {}))
//                .print("key:");




        env.execute();


    }

}
